package com.huawei.hiveFunctions;

import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDTF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;

import java.util.*;

/**
 * 自定义的UDTF函数
 * 实现功能：解析爬虫数据，从数据读取产品ID，产品名称，价格等
 * 创建一个自定义函数UDTF，来实现一个输入多次输出的函数，我们提供好一些数据，通过这些数据的清洗，我们可以实现在数据中读取产品ID，产品名称，价格等
 */

public class UDTFCase extends GenericUDTF {
    @Override
    public StructObjectInspector initialize(StructObjectInspector argOIs) throws UDFArgumentException {
        if(argOIs.getAllStructFieldRefs().size() != 1){
            throw new UDFArgumentException("参数异常");
        }
        ArrayList<String> filedName =new ArrayList<String>();
        ArrayList<ObjectInspector> filedOIs = new ArrayList<ObjectInspector>();
        filedName.add("id");
        filedName.add("name");
        filedName.add("price");
        filedOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        filedOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        filedOIs.add(PrimitiveObjectInspectorFactory.javaStringObjectInspector);
        return ObjectInspectorFactory.getStandardStructObjectInspector(filedName,filedOIs);
    }

    public void process(Object[] objects) throws HiveException {
        if(objects == null || objects.length !=1){
            return;
        }

        //只有一个参数的情况
        String line = objects[0].toString();
        //解析
        Map<String,String> map = transfoerContent2Map(line);
        List<String> result = new ArrayList<String>();
        result.add(map.get("p_id"));
        result.add(map.get("p_name"));
        result.add(map.get("price"));
        super.forward(result.toArray(new String[0]));
    }

    public void close() throws HiveException {
        //nothing
        super.forward(new String[] {"12345689","close","123"});
    }

    /*
    转换字符串为map对象
     */
    static Map<String, String> transfoerContent2Map(String content) {
        Map<String, String> map = new HashMap<String, String>();
        int i = 0;
        String key = "";
        StringTokenizer tokenizer = new StringTokenizer(content, "({|}|\"|:|,)");
        while (tokenizer.hasMoreTokens()) {
            if (++i % 2 == 0) {
                // 当前的值是value
                map.put(key, tokenizer.nextToken());
            } else {
                // 当前的值是key
                key = tokenizer.nextToken();
            }
        }
        return map;
    }
}